package spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot;

import com.google.common.collect.Lists;
import com.google.gson.Gson;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import spring.cloud.tasks.tasks_manager.tasks_manager.node.Node;
import spring.cloud.tasks.tasks_manager.tasks_manager.service.AllotTaskItemContentService;
import spring.cloud.tasks.tasks_manager.tasks_manager.service.AllotTaskItemService;
import spring.cloud.tasks.tasks_manager.tasks_manager.service.ReportAlarmService;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.executor.Executor;
import spring.cloud.tasks.tasks_manager.tasks_manager.task_item_allot.task.TaskItem;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;

@Slf4j
@Setter
@Getter
public abstract class TaskItemAllotTask implements Runnable {

    private static final String NAME_ENABLE_JOB_BASED_SHARDING = "VIP_SATURN_ENABLE_JOB_BASED_SHARDING";
    public static boolean ENABLE_TASK_BASED_SHARDING = true;

    static {
        String enableJobAverage = System.getProperty(NAME_ENABLE_JOB_BASED_SHARDING, System.getenv(NAME_ENABLE_JOB_BASED_SHARDING));
        if (StringUtils.isNotBlank(enableJobAverage)) {
            ENABLE_TASK_BASED_SHARDING = Boolean.parseBoolean(enableJobAverage);
        }
        log.info("ENABLE_JOB_BASED_SHARDING is {}", ENABLE_TASK_BASED_SHARDING);
    }

    protected AllotTaskItemService allotTaskItemService;

    protected CuratorFramework curatorFramework;

    protected AllotTaskItemContentService allotTaskItemContentService;

    protected ExecutorService executorService;

    protected ReportAlarmService reportAlarmService;

    public TaskItemAllotTask(AllotTaskItemService allotTaskItemService) {
        this.allotTaskItemService = allotTaskItemService;
        this.curatorFramework = allotTaskItemService.getCuratorFramework();
        this.allotTaskItemContentService = allotTaskItemService.getAllotTaskItemContentService();
        this.executorService = allotTaskItemService.getExecutorService();
        this.reportAlarmService = allotTaskItemService.getReportAlarmService();
    }

    protected abstract void logStartInfo();

    /**
     * Special enable jobs that need to be notified prior, not consider whether whose shards are changed.
     * By default, notify enable jobs whose shards are changed.
     */
    protected List<String> notifyEnableTaskIdListPrior() {
        return null;
    }

    @Override
    public void run() {
        logStartInfo();
        boolean isTasksFullTaskItemTask = this instanceof TasksFullTaskItemTask;
        try {
            // 如果当前变为非leader，则直接返回
            if (!allotTaskItemService.isLeadershipOnly()) {
                return;
            }

            // 如果需要全量分片，且当前线程不是全量分片线程，则直接返回，没必要做分片
            if (allotTaskItemService.isNeedAllSharding() && !isTasksFullTaskItemTask) {
                log.info("the {} will be ignored, because there will be {}", this.getClass().getSimpleName(), TasksFullTaskItemTask.class.getSimpleName());
                return;
            }

            List<String> taskIdList = getTaskIdList();
            List<String> enableTaskIdList = getEnableTaskIdList(taskIdList);
            //
            List<Executor> lastOnlineExecutorList = getLastOnlineExecutorList();
            List<Executor> customLastOnlineExecutorList = customLastOnlineExecutorList();
            //
            List<Executor> onlineExecutorList = customLastOnlineExecutorList == null ? copyOnlineExecutorList(lastOnlineExecutorList) : customLastOnlineExecutorList;
            List<Executor> onlineOwnTrafficExecutorList = getOnlineOwnTrafficExecutorList(onlineExecutorList);
            //
            List<TaskItem> taskItemList = new ArrayList<TaskItem>();
            // 摘取
            if (pick(taskIdList, enableTaskIdList, taskItemList, onlineExecutorList, onlineOwnTrafficExecutorList)) {
                // 放回
                putBack(enableTaskIdList, taskItemList, onlineExecutorList, onlineOwnTrafficExecutorList);
                // 如果当前变为非leader，则返回
                if (!allotTaskItemService.isLeadershipOnly()) {
                    return;
                }
                // 持久化分片结果
                if (executorListContentIsChanged(lastOnlineExecutorList, onlineExecutorList)) {
                    allotTaskItemContentService.persistExecutorList(onlineExecutorList);
                }
                // notify the shards-changed jobs of all enable jobs.
                Map<String, Map<String, List<Integer>>> enableTaskIdToExecutorIdToItemIndexListMapMap = getEnableTaskIdToExecutorIdToItemIndexListMapMap(isTasksFullTaskItemTask, enableTaskIdList, lastOnlineExecutorList, onlineExecutorList);
                allotTaskItemContentService.persistTaskIdToExecutorIdToItemIndexListMapMap(enableTaskIdToExecutorIdToItemIndexListMapMap);
                // sharding count ++
                increaseShardingCount();
            }
        } catch (InterruptedException e) {
            log.info("{}-{} {} is interrupted", allotTaskItemService.getNamespace(),
                    allotTaskItemService.getHostValue(), this.getClass().getSimpleName());
            Thread.currentThread().interrupt();
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
            if (!isTasksFullTaskItemTask) { // 如果当前不是全量分片，则需要全量分片来拯救异常
                allotTaskItemService.setNeedAllSharding(true);
                allotTaskItemService.shardingCountIncrementAndGet();
                executorService.submit(new TasksFullTaskItemTask(allotTaskItemService));
            } else { // 如果当前是全量分片，则告警并关闭当前服务，重选leader来做事情
                raiseAlarm();
                shutdownAllotTaskItemService();
            }
        } finally {
            if (isTasksFullTaskItemTask) { // 如果是全量分片，不再进行全量分片
                allotTaskItemService.setNeedAllSharding(false);
            }
            allotTaskItemService.shardingCountDecrementAndGet();
        }
    }

    /**
     * 获取该域下的所有作业
     */
    private List<String> getTaskIdList() throws Exception {

        if (curatorFramework.checkExists().forPath(Node.TASKS_NODE_PATH) == null) {
            curatorFramework.create().creatingParentsIfNeeded().forPath(Node.TASKS_NODE_PATH);
        }
        List<String> taskIdList = new ArrayList<>();
        List<String> $taskIdList = curatorFramework.getChildren().forPath(Node.TASKS_NODE_PATH);
        if ($taskIdList != null) {
            taskIdList.addAll($taskIdList);
        }
        return taskIdList;
    }

    /**
     * 获取该域下的所有enable的作业
     */
    protected List<String> getEnableTaskIdList(List<String> taskIdList) throws Exception {
        List<String> enableTaskIdList = new ArrayList<>();
        for (int i = 0; i < taskIdList.size(); i++) {
            String taskId = taskIdList.get(i);
            if (curatorFramework.checkExists().forPath(Node.getJobConfigEnableNodePath(taskId)) != null) {
                byte[] bytes = curatorFramework.getData().forPath(Node.getJobConfigEnableNodePath(taskId));
                if (bytes != null && Boolean.parseBoolean(new String(bytes, StandardCharsets.UTF_8.name()))) {
                    enableTaskIdList.add(taskId);
                }
            }
        }
        return enableTaskIdList;
    }

    /**
     * 获取Executor集合，默认从sharding/content获取
     */
    private List<Executor> getLastOnlineExecutorList() throws Exception {
        return allotTaskItemContentService.getExecutorList();
    }

    /**
     * Custom the lastOnlineExecutorList, attention, cannot return null
     */
    protected List<Executor> customLastOnlineExecutorList() throws Exception {
        return null;
    }

    private List<Executor> copyOnlineExecutorList(List<Executor> oldOnlineExecutorList) {
        List<Executor> newOnlineExecutorList = new ArrayList<>();
        for (Executor oldExecutor : oldOnlineExecutorList) {
            Executor newExecutor = new Executor();
            newExecutor.setExecutorId(oldExecutor.getExecutorId());
            newExecutor.setIp(oldExecutor.getIp());
            newExecutor.setLostTraffic(oldExecutor.isLostTraffic());
            newExecutor.setTotalLoadLevel(oldExecutor.getTotalLoadLevel());
            if (oldExecutor.getTaskIdList() != null) {
                newExecutor.setTaskIdList(new ArrayList<String>());
                for (String taskId : oldExecutor.getTaskIdList()) {
                    newExecutor.getTaskIdList().add(taskId);
                }
            }
            if (oldExecutor.getTaskItemList() != null) {
                newExecutor.setTaskItemList(new ArrayList<TaskItem>());
                for (TaskItem oldTaskItem : oldExecutor.getTaskItemList()) {
                    TaskItem taskItem = new TaskItem();
                    taskItem.setTaskId(oldTaskItem.getTaskId());
                    taskItem.setTaskItemIndex(oldTaskItem.getTaskItemIndex());
                    taskItem.setLoadLevel(oldTaskItem.getLoadLevel());
                    newExecutor.getTaskItemList().add(taskItem);
                }
            }
            newOnlineExecutorList.add(newExecutor);
        }
        return newOnlineExecutorList;
    }

    private List<Executor> getOnlineOwnTrafficExecutorList(List<Executor> onlineExecutorList) {
        List<Executor> onlineOwnTrafficExecutorList = new ArrayList<>();
        for (Executor executor : onlineExecutorList) {
            if (!executor.isLostTraffic()) {
                onlineOwnTrafficExecutorList.add(executor);
            }
        }
        return onlineOwnTrafficExecutorList;
    }

    /**
     * 摘取
     *
     * @param taskIdList                   该域下所有作业
     * @param enableTaskIdList             该域下所有启用的作业
     * @param taskItemList                 默认为空集合
     * @param onlineExecutorList           默认为当前存储的数据，如果不想使用存储数据，请重写{@link #customLastOnlineExecutorList()}}方法
     * @param onlineOwnTrafficExecutorList lastOnlineExecutorList中所有noTraffic为false的Executor，注意Executor是同一个对象
     * @return true摘取成功；false摘取失败，不需要继续下面的逻辑
     */
    protected abstract boolean pick(List<String> taskIdList,//
                                    List<String> enableTaskIdList,//
                                    List<TaskItem> taskItemList,//
                                    List<Executor> onlineExecutorList,//
                                    List<Executor> onlineOwnTrafficExecutorList) throws Exception;

    protected void putBack(List<String> enableTaskIdList, List<TaskItem> taskItemList, List<Executor> onlineExecutorList, List<Executor> onlineOwnTrafficExecutorList) throws Exception {
        if (onlineExecutorList.isEmpty()) {
            log.warn("Unnecessary to put shards back to executors balanced because of no executor");
            return;
        }

        sortTaskItemList(taskItemList);

        // 获取所有非容器的executors
        List<Executor> noneContainerOnlineOwnTrafficExecutorList = getNoneContainerOnlineOwnTrafficExecutorList(onlineOwnTrafficExecutorList);

        // 获取shardList中的作业能够被接管的executors
        Map<String, List<Executor>> taskIdToNoneContainerOnlineOwnTrafficExecutorListMap = new HashMap<>();
        Map<String, List<Executor>> taskIdToOnlineOwnTrafficExecutorListMap = new HashMap<>();

        // 是否配置优先节点的作业的映射
        Map<String, Boolean> taskIdToPreferExecutorIdListIsConfiguredMap = new HashMap<String, Boolean>();
        // 优先节点的作业的映射
        Map<String, List<String>> taskIdToPreferExecutorIdListConfiguredMap = new HashMap<String, List<String>>();
        // 是否使用非优先节点的作业的映射
        Map<String, Boolean> taskIdToUseDisPreferExecutorIdListMap = new HashMap<String, Boolean>();

        Iterator<TaskItem> iterator = taskItemList.iterator();
        while (iterator.hasNext()) {
            String taskId = iterator.next().getTaskId();
            putIfAbsent(taskIdToOnlineOwnTrafficExecutorListMap, taskId, getExecutorListByTaskId(onlineOwnTrafficExecutorList, taskId));
            putIfAbsent(taskIdToNoneContainerOnlineOwnTrafficExecutorListMap, taskId, getExecutorListByTaskId(noneContainerOnlineOwnTrafficExecutorList, taskId));
            //
            putIfAbsent(taskIdToPreferExecutorIdListIsConfiguredMap, taskId, preferExecutorIdListIsConfigured(taskId));
            putIfAbsent(taskIdToPreferExecutorIdListConfiguredMap, taskId, getPreferExecutorIdListConfigured(taskId));
            //
            putIfAbsent(taskIdToUseDisPreferExecutorIdListMap, taskId, useDisPreferExecutorIdList(taskId));
        }

        // 整体算法放回算法：拿取Shard，放进负荷最小的executor
        // 2、放回配置了preferList的Shard
        putBackWithPreferList(taskItemList, taskIdToOnlineOwnTrafficExecutorListMap, taskIdToPreferExecutorIdListIsConfiguredMap, taskIdToPreferExecutorIdListConfiguredMap, taskIdToUseDisPreferExecutorIdListMap);

        // 3、放回没有配置preferList的Shard
        putBackWithoutPreferList(taskItemList, taskIdToNoneContainerOnlineOwnTrafficExecutorListMap);
    }

    /**
     * 按照loadLevel降序排序，如果loadLevel相同，按照作业名降序排序
     */
    protected void sortTaskItemList(List<TaskItem> taskItemList) {
        Collections.sort(taskItemList, new Comparator<TaskItem>() {
            @Override
            public int compare(TaskItem o1, TaskItem o2) {
                int loadLevelSub = o2.getLoadLevel() - o1.getLoadLevel();
                return loadLevelSub == 0 ? o2.getTaskId().compareTo(o1.getTaskId()) : loadLevelSub;
            }
        });
    }

    private List<Executor> getNoneContainerOnlineOwnTrafficExecutorList(List<Executor> lastOnlineExecutorList) throws Exception {
        // if CONTAINER_ALIGN_WITH_PHYSICAL = false, return all executors; otherwise, return all non-container executors.
        if (AllotTaskItemService.CONTAINER_ALIGN_WITH_PHYSICAL) {
            return lastOnlineExecutorList;
        }
        List<Executor> executorList = new ArrayList<>();
        for (int i = 0; i < lastOnlineExecutorList.size(); i++) {
            Executor executor = lastOnlineExecutorList.get(i);
            String executorId = executor.getExecutorId();
            if (!executorId.startsWith("@")) {
                executorList.add(executor);
            }
        }
        return executorList;
    }

    private <T> void putIfAbsent(Map<String, T> map, String key, T value) {
        if (!map.containsKey(key)) {
            map.put(key, value);
        }
    }

    protected List<Executor> getExecutorListByTaskId(List<Executor> executorList, String taskId) {
        List<Executor> $executorList = new ArrayList<>();
        for (int i = 0; i < executorList.size(); i++) {
            Executor executor = executorList.get(i);
            List<String> taskIdList = executor.getTaskIdList();
            if (taskIdList != null && taskIdList.contains(taskId)) {
                $executorList.add(executor);
            }
        }
        return $executorList;
    }

    protected boolean preferExecutorIdListIsConfigured(String taskId) throws Exception {
        if (curatorFramework.checkExists().forPath(Node.getTaskConfigPreferListNodePath(taskId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(Node.getTaskConfigPreferListNodePath(taskId));
            if (bytes != null) {
                return new String(bytes, StandardCharsets.UTF_8.name()).trim().length() > 0;
            }
        }
        return false;
    }

    /**
     * 获取配置态的preferList，即使配置的executor不存在，也会返回。 特别的是，对于docker task，如果存在，才去解析出executor列表。
     */
    protected List<String> getPreferExecutorIdListConfigured(String taskId) throws Exception {
        List<String> preferExecutorIdList = new ArrayList<>();
        if (curatorFramework.checkExists().forPath(Node.getTaskConfigPreferListNodePath(taskId)) != null) {
            byte[] bytes = curatorFramework.getData().forPath(Node.getTaskConfigPreferListNodePath(taskId));
            if (bytes != null) {
                List<String> executorIdList = getExecutorIdList();
                String[] preferExecutorIds = new String(bytes, StandardCharsets.UTF_8.name()).split(",");
                for (String preferExecutorId : preferExecutorIds) {
                    if (executorIdList.contains(preferExecutorId) && !preferExecutorIdList.contains(preferExecutorId)) {
                        preferExecutorIdList.add(preferExecutorId);
                    }
                }
            }
        }
        return preferExecutorIdList;
    }

    private List<String> getExecutorIdList() throws Exception {
        List<String> executorIdList = new ArrayList<>();
        if (curatorFramework.checkExists().forPath(Node.getExecutorsNodePath()) != null) {
            List<String> $executorIdList = curatorFramework.getChildren().forPath(Node.getExecutorsNodePath());
            if ($executorIdList != null) {
                executorIdList.addAll($executorIdList);
            }
        }
        return executorIdList;
    }

    /**
     * 是否使用非preferList:
     * 1、存在结点，并且该结点值为false，返回false；
     * 2、其他情况，返回true
     * 默认设置为true
     */
    protected boolean useDisPreferExecutorIdList(String taskId) throws Exception {
        String useDisPreferExecutorIdListNodePath = Node.getTaskConfigUseDispreferListNodePath(taskId);
        if (curatorFramework.checkExists().forPath(useDisPreferExecutorIdListNodePath) != null) {
            byte[] bytes = curatorFramework.getData().forPath(useDisPreferExecutorIdListNodePath);
            return bytes == null || Boolean.parseBoolean(new String(bytes, StandardCharsets.UTF_8.name()));
        }
        return true;
    }

    private void putBackWithPreferList(List<TaskItem> taskItemList,//
                                       Map<String, List<Executor>> taskIdToOnlineOwnTrafficExecutorListMap,//
                                       Map<String, Boolean> taskIdToPreferExecutorIdListIsConfiguredMap,//
                                       Map<String, List<String>> taskIdToPreferExecutorIdListConfiguredMap,//
                                       Map<String, Boolean> taskIdToUseDisPreferExecutorIdListMap) {//
        Iterator<TaskItem> iterator = taskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            String taskId = taskItem.getTaskId();
            // preferList为空不能作为判断是否配置preferList的依据，比如说配置了容器资源，但是全部下线了。
            if (taskIdToPreferExecutorIdListIsConfiguredMap.get(taskId)) {
                List<String> preferExecutorIdListConfigured = taskIdToPreferExecutorIdListConfiguredMap.get(taskId);
                List<Executor> preferExecutorList = getPreferExecutorList(taskIdToOnlineOwnTrafficExecutorListMap, taskId, preferExecutorIdListConfigured);
                // 如果preferList的Executor都offline，则放回到全部online的Executor中某一个。如果是这种情况，则后续再操作，避免不均衡的情况
                // 如果存在preferExecutor，择优放回
                if (!preferExecutorList.isEmpty()) {
                    Executor executor = ENABLE_TASK_BASED_SHARDING ? getExecutorWithMinTaskTotalTaskLoadLevelOnExecutor(preferExecutorList, taskId) : getExecutorWithMinTotalLoadLevel(preferExecutorList);
                    addTaskItemToExecutor(taskItem, executor);
                    iterator.remove();
                } else { // 如果不存在preferExecutor
                    // 如果“只使用preferExecutor”，则丢弃；否则，等到后续（在第3步）进行放回操作，避免不均衡的情况
                    if (!taskIdToUseDisPreferExecutorIdListMap.get(taskId)) {
                        iterator.remove();
                    }
                }
            }
        }
    }

    private List<Executor> getPreferExecutorList(Map<String, List<Executor>> taskIdToOnlineOwnTrafficExecutorListMap, String taskId, List<String> preferExecutorIdListConfigured) {
        List<Executor> preferExecutorList = new ArrayList<>();
        List<Executor> onlineOwnTrafficExecutorList = taskIdToOnlineOwnTrafficExecutorListMap.get(taskId);
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            Executor executor = onlineOwnTrafficExecutorList.get(i);
            if (preferExecutorIdListConfigured.contains(executor.getExecutorId())) {
                preferExecutorList.add(executor);
            }
        }
        return preferExecutorList;
    }

    private void putBackWithoutPreferList(List<TaskItem> taskItemList, Map<String, List<Executor>> taskIdToNoneContainerOnlineOwnTrafficExecutorListMap) {
        Iterator<TaskItem> iterator = taskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            List<Executor> executorList = taskIdToNoneContainerOnlineOwnTrafficExecutorListMap.get(taskItem.getTaskId());
            Executor executor = ENABLE_TASK_BASED_SHARDING ? getExecutorWithMinTaskTotalTaskLoadLevelOnExecutor(executorList, taskItem.getTaskId()) : getExecutorWithMinTotalLoadLevel(executorList);
            addTaskItemToExecutor(taskItem, executor);
            iterator.remove();
        }
    }

    /**
     * 获取该作业负荷最小的executor，如果相同，那么取所有作业负荷最小的executor
     */
    private Executor getExecutorWithMinTaskTotalTaskLoadLevelOnExecutor(List<Executor> executorList, String taskId) {
        Executor minTaskTotalTaskLoadLevelExecutor = null;
        int minTotalLoadLevel = 0;
        for (int i = 0; i < executorList.size(); i++) {
            Executor executor = executorList.get(i);
            int taskTotalTaskLoadLevel = getTaskTotalTaskLoadLevelOnExecutor(executor, taskId);
            if (minTaskTotalTaskLoadLevelExecutor == null || minTotalLoadLevel > taskTotalTaskLoadLevel || minTotalLoadLevel == taskTotalTaskLoadLevel && minTaskTotalTaskLoadLevelExecutor.getTotalLoadLevel() > executor.getTotalLoadLevel()) {
                minTaskTotalTaskLoadLevelExecutor = executor;
                minTotalLoadLevel = taskTotalTaskLoadLevel;
            }
        }
        return minTaskTotalTaskLoadLevelExecutor;
    }

    /**
     * 计算指定作业在特定executor上的负荷
     */
    private int getTaskTotalTaskLoadLevelOnExecutor(Executor executor, String taskId) {
        int totalTaskLoadLevel = 0;
        List<TaskItem> taskItemList = executor.getTaskItemList();
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem taskItem = taskItemList.get(i);
            if (taskId.equals(taskItem.getTaskId())) {
                totalTaskLoadLevel += taskItem.getLoadLevel();
            }
        }
        return totalTaskLoadLevel;
    }

    private Executor getExecutorWithMinTotalLoadLevel(List<Executor> executorList) {
        Executor minLoadLevelExecutor = null;
        for (int i = 0; i < executorList.size(); i++) {
            Executor executor = executorList.get(i);
            if (minLoadLevelExecutor == null || minLoadLevelExecutor.getTotalLoadLevel() > executor.getTotalLoadLevel()) {
                minLoadLevelExecutor = executor;
            }
        }
        return minLoadLevelExecutor;
    }

    private void addTaskItemToExecutor(TaskItem taskItem, Executor executor) {
        if (executor != null) {
            if (isIn(executor.getTaskItemList(), taskItem)) {
                log.error("The shard({}-{}) is running in the executor of {}, cannot be put again", taskItem.getTaskId(), taskItem.getTaskItemIndex(), executor.getExecutorId());
            } else {
                executor.getTaskItemList().add(taskItem);
                executor.setTotalLoadLevel(executor.getTotalLoadLevel() + taskItem.getLoadLevel());
            }
        } else {
            log.info("No executor to take over the shard: {}-{}", taskItem.getTaskId(), taskItem.getTaskItemIndex());
        }
    }

    protected boolean isIn(List<TaskItem> taskItemList, TaskItem taskItem) {
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem $taskItem = taskItemList.get(i);
            if ($taskItem.getTaskId().equals(taskItem.getTaskId()) && $taskItem.getTaskItemIndex() == taskItem.getTaskItemIndex()) {
                return true;
            }
        }
        return false;
    }

    private void raiseAlarm() {
        if (reportAlarmService != null) {
            try {
                reportAlarmService.allShardingError(allotTaskItemService.getNamespace(), allotTaskItemService.getHostValue());
            } catch (Throwable t) {
                log.error(t.getMessage(), t);
            }
        }
    }

    private void shutdownAllotTaskItemService() {
        try {
            allotTaskItemService.shutdownInner(false);
        } catch (InterruptedException e) {
            log.info("{}-{} {}-shutdownInner is interrupted", allotTaskItemService.getNamespace(), allotTaskItemService.getHostValue(), this.getClass().getSimpleName());
            Thread.currentThread().interrupt();
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
        }
    }


    private boolean executorListContentIsChanged(List<Executor> oldOnlineExecutorList, List<Executor> newOnlineExecutorList) {
        return !(new Gson()).toJson(oldOnlineExecutorList).equals((new Gson()).toJson((newOnlineExecutorList)));
    }

    /**
     * Get the jobs, that are enabled, and whose shards are changed. Specially, return all enabled jobs when the current
     * thread is all-shard-task
     * Return the jobs and their shardContent.
     */
    private Map<String, Map<String, List<Integer>>> getEnableTaskIdToExecutorIdToItemIndexListMapMap(//
                                                                                                     boolean isTasksFullTaskItemTask,//
                                                                                                     List<String> enableTaskIdList,//
                                                                                                     List<Executor> lastOnlineExecutorList,//
                                                                                                     List<Executor> onlineExecutorList) {
        Map<String, Map<String, List<Integer>>> taskIdToExecutorIdToItemIndexListMapMap = new HashMap<>();
        if (isTasksFullTaskItemTask) {
            for (String taskId : enableTaskIdList) {
                Map<String, List<Integer>> onlineExecutorList$executorIdToItemIndexListMap = allotTaskItemContentService.getExecutorIdToItemIndexListMap(onlineExecutorList, taskId);//按照新的计算
                taskIdToExecutorIdToItemIndexListMapMap.put(taskId, onlineExecutorList$executorIdToItemIndexListMap);
            }
            return taskIdToExecutorIdToItemIndexListMapMap;
        }
        List<String> enableTaskIdListPrior = notifyEnableTaskIdListPrior();
        for (String taskId : enableTaskIdList) {
            Map<String, List<Integer>> onlineExecutorList$executorIdToItemIndexListMap = allotTaskItemContentService.getExecutorIdToItemIndexListMap(onlineExecutorList, taskId);
            // notify prior jobs that are in all enable jobs
            if (enableTaskIdListPrior != null && enableTaskIdListPrior.contains(taskId)) {
                taskIdToExecutorIdToItemIndexListMapMap.put(taskId, onlineExecutorList$executorIdToItemIndexListMap);
                continue;
            }
            Map<String, List<Integer>> lastOnlineExecutorList$executorIdToItemIndexListMap = allotTaskItemContentService.getExecutorIdToItemIndexListMap(lastOnlineExecutorList, taskId);
            // just compare whether or not contains the same executorName, and it's shardList
            boolean isChanged = false;
            //先以lastOnlineExecutorList$executorIdToItemIndexListMap为参考进行对比
            Iterator<Entry<String, List<Integer>>> iterator = lastOnlineExecutorList$executorIdToItemIndexListMap.entrySet().iterator();
            wl_loop:
            while (iterator.hasNext()) {
                Entry<String, List<Integer>> next = iterator.next();
                String executorId = next.getKey();
                if (!onlineExecutorList$executorIdToItemIndexListMap.containsKey(executorId)) {
                    isChanged = true;
                    break;
                }
                List<Integer> taskItemIndexList = next.getValue();
                List<Integer> onlineExecutorList$TaskItemIndexList = onlineExecutorList$executorIdToItemIndexListMap.get(executorId);
                if ((taskItemIndexList == null && onlineExecutorList$TaskItemIndexList != null) || (taskItemIndexList != null && onlineExecutorList$TaskItemIndexList == null)) {
                    isChanged = true;
                    break;
                }
                if (taskItemIndexList == null || onlineExecutorList$TaskItemIndexList == null) {
                    continue;
                }
                for (Integer taskItemIndex : taskItemIndexList) {
                    if (!onlineExecutorList$TaskItemIndexList.contains(taskItemIndex)) {
                        isChanged = true;
                        break wl_loop;
                    }
                }
            }
            if (!isChanged) {
                //再以onlineExecutorList$executorIdToItemIndexListMap为参考进行对比
                Iterator<Entry<String, List<Integer>>> $iterator = onlineExecutorList$executorIdToItemIndexListMap.entrySet().iterator();
                while ($iterator.hasNext()) {
                    Entry<String, List<Integer>> next = $iterator.next();
                    String executorId = next.getKey();
                    if (!lastOnlineExecutorList$executorIdToItemIndexListMap.containsKey(executorId)) {
                        isChanged = true;
                        break;
                    }
                    List<Integer> taskItemIndexList = next.getValue();
                    List<Integer> lastOnlineExecutorList$TaskItemIndexList = lastOnlineExecutorList$executorIdToItemIndexListMap.get(executorId);
                    if ((taskItemIndexList == null && lastOnlineExecutorList$TaskItemIndexList != null) || (taskItemIndexList != null && lastOnlineExecutorList$TaskItemIndexList == null)) {
                        isChanged = true;
                        break;
                    }

                    if (taskItemIndexList == null || lastOnlineExecutorList$TaskItemIndexList == null) {
                        continue;
                    }

                    if (taskItemIndexAdded(lastOnlineExecutorList$TaskItemIndexList, taskItemIndexList)) {
                        isChanged = true;
                        break;
                    }
                }
            }

            if (isChanged) {
                taskIdToExecutorIdToItemIndexListMapMap.put(taskId, onlineExecutorList$executorIdToItemIndexListMap);
            }
        }
        return taskIdToExecutorIdToItemIndexListMapMap;
    }

    private boolean taskItemIndexAdded(List<Integer> oldTaskItemIndexList, List<Integer> newTaskItemIndexList) {
        for (Integer newTaskItemIndex : newTaskItemIndexList) {
            if (!oldTaskItemIndexList.contains(newTaskItemIndex)) {
                return true;
            }
        }
        return false;
    }

    private void increaseShardingCount() throws Exception {
        Integer shardingCount = 1;
        if (curatorFramework.checkExists().forPath(Node.SHARDING_COUNT_PATH) != null) {
            byte[] bytes = curatorFramework.getData().forPath(Node.SHARDING_COUNT_PATH);
            if (bytes != null) {
                try {
                    shardingCount = Integer.parseInt(new String(bytes, StandardCharsets.UTF_8.name())) + 1;
                } catch (NumberFormatException e) {
                    log.error("parse shardingCount error", e);
                    throw new RuntimeException(e);
                }
            }
            curatorFramework.setData().forPath(Node.SHARDING_COUNT_PATH, shardingCount.toString().getBytes(StandardCharsets.UTF_8.name()));
        } else {
            curatorFramework.create().creatingParentsIfNeeded().forPath(Node.SHARDING_COUNT_PATH, shardingCount.toString().getBytes(StandardCharsets.UTF_8.name()));
        }
    }
    //##<!-我是优美的功能模块分割线->##


    protected boolean getExecutorLostTraffic(String executorId) throws Exception {
        return curatorFramework.checkExists().forPath(Node.getExecutorLostTrafficNodePath(executorId)) != null;
    }


    //##<!-我是优美的功能模块分割线->##
    protected List<TaskItem> createTaskItemList(String taskId) throws Exception {
        List<TaskItem> taskItemList = new ArrayList<>();
        int totalTaskItemCount = getTotalTaskItemCount(taskId);
        int loadLevel = getLoadLevel(taskId);
        // 新建shardingTotalCount数量的分片
        taskItemList.addAll(createTaskItemList(taskId, totalTaskItemCount, loadLevel));
        return taskItemList;
    }

    protected int getTotalTaskItemCount(String taskId) throws Exception {
        String totalTaskItemCountNodePath = Node.getTaskConfigTotalTaskItemCountNodePath(taskId);
        if (curatorFramework.checkExists().forPath(totalTaskItemCountNodePath) != null) {
            byte[] bytes = curatorFramework.getData().forPath(totalTaskItemCountNodePath);
            if (bytes != null) {
                try {
                    int totalTaskItemCount = Integer.parseInt(new String(bytes, StandardCharsets.UTF_8.name()));
                    return totalTaskItemCount;
                } catch (NumberFormatException e) {
                    log.error("parse shardingTotalCount error, will use the default value", e);
                    throw new RuntimeException(e);
                }
            } else {
                throw new RuntimeException();
            }
        } else {
            throw new RuntimeException();
        }
    }

    protected int getLoadLevel(String taskId) throws Exception {

        String taskConfigLoadLevelNodePath = Node.getTaskConfigLoadLevelNodePath(taskId);
        if (curatorFramework.checkExists().forPath(taskConfigLoadLevelNodePath) != null) {
            byte[] bytes = curatorFramework.getData().forPath(taskConfigLoadLevelNodePath);
            if (bytes != null) {
                try {
                    int loadLevel = Integer.parseInt(new String(bytes, StandardCharsets.UTF_8.name()));
                    return loadLevel;
                } catch (NumberFormatException e) {
                    log.error("parse loadLevel error, will use the default value", e);
                    throw new RuntimeException(e);
                }

            } else {
                throw new RuntimeException();
            }
        } else {
            throw new RuntimeException();
        }
    }

    private List<TaskItem> createTaskItemList(String taskId, int totalTaskItemCount, int loadLevel) {
        List<TaskItem> taskItemList = new ArrayList<>();
        for (int taskItemIndex = 0; taskItemIndex < totalTaskItemCount; taskItemIndex++) {
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskId(taskId);
            taskItem.setTaskItemIndex(taskItemIndex);
            taskItem.setLoadLevel(loadLevel);
            taskItemList.add(taskItem);
        }
        return taskItemList;
    }

    /**
     * 刷新onlineExecutorList taskIdList
     */
    protected boolean refreshExecutorTaskIdList(List<Executor> onlineExecutorList, String taskId) throws Exception {
        boolean fixed = false;
        for (int i = 0; i < onlineExecutorList.size(); i++) {
            Executor executor = onlineExecutorList.get(i);
            if (executor.getTaskIdList() == null) {
                executor.setTaskIdList(new ArrayList<String>());
            }
            List<String> taskIdList = executor.getTaskIdList();
            String taskExecutorStatusNodePath = Node.getTaskExecutorStatusNodePath(taskId, executor.getExecutorId());
            if (curatorFramework.checkExists().forPath(taskExecutorStatusNodePath) != null) {
                if (!taskIdList.contains(taskId)) {
                    taskIdList.add(taskId);
                    fixed = true;
                }
            } else {
                if (taskIdList.contains(taskId)) {
                    taskIdList.remove(taskId);
                    fixed = true;
                }
            }
        }
        return fixed;
    }

    /**
     * 移除特定作业的shard
     */
    protected List<TaskItem> removeTaskItemList(List<Executor> onlineOwnTrafficExecutorList, String taskId) {
        List<TaskItem> removedTaskItemList = Lists.newArrayList();
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            Executor executor = onlineOwnTrafficExecutorList.get(i);
            Iterator<TaskItem> iterator = executor.getTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {
                    executor.setTotalLoadLevel(executor.getTotalLoadLevel() - taskItem.getLoadLevel());
                    iterator.remove();
                    removedTaskItemList.add(taskItem);
                }
            }
        }
        return removedTaskItemList;
    }


    //##<!-我是优美的功能模块分割线->##


}
